package com.touwowo.four;

import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.MessageProperties;
import com.touwowo.utils.RabbitMqUtils;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

public class ConfirmMessage {
    private static final int MESSAGE_NUM = 1000;

    public static void main(String[] args) throws Exception {
        //confirmMessageMuti();
        confirmMessageAysnc();
    }
    public static void confirmMessageSingal() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        //设置发布确认
        channel.confirmSelect();
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);

        long start = System.currentTimeMillis();

        for (int i = 0; i < MESSAGE_NUM; i++) {
            String message = i + "";
            channel.basicPublish("",queueName, null,message.getBytes(StandardCharsets.UTF_8));
            channel.waitForConfirms();
        }

        long end = System.currentTimeMillis();
        //单个确认发布 1000 条消息的耗时为 47443 ms
        System.out.println("单个确认发布 " + MESSAGE_NUM + " 条消息的耗时为 "+ (end - start) +" ms");
    }

    public static void confirmMessageMuti() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        //设置发布确认
        channel.confirmSelect();
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);
        long start = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_NUM; i++) {
            String message = i + "";
            channel.basicPublish("",queueName, null,message.getBytes(StandardCharsets.UTF_8));
            if((i+1)%100 == 0){
                channel.waitForConfirms();
            }
        }
        long end = System.currentTimeMillis();
        // 批量确认发布 1000 条消息的耗时为 624 ms
        System.out.println("批量确认发布 " + MESSAGE_NUM + " 条消息的耗时为 "+ (end - start) +" ms");
    }

    public static void confirmMessageAysnc() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        //设置发布确认
        channel.confirmSelect();
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);

        ConcurrentSkipListMap<Long,String> outstandmMap = new ConcurrentSkipListMap<>();

        ConfirmCallback confirmCallback = (deliveryTag,multiple) ->{
            if(multiple){
                ConcurrentNavigableMap<Long, String> confirms = outstandmMap.headMap(deliveryTag);
                confirms.clear();
            }else {
                outstandmMap.remove(deliveryTag);
            }
            System.out.println("收到消息的内容："+deliveryTag);
        };
        ConfirmCallback cancelCallback = (deliveryTag,multiple) ->{
            String s = outstandmMap.get(deliveryTag);
            System.out.println("未确认消息的编号："+s);
        };
        channel.addConfirmListener(confirmCallback,cancelCallback);
        long start = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_NUM; i++) {
            String message = i + "";
            channel.basicPublish("",queueName, null,message.getBytes(StandardCharsets.UTF_8));
            long nextPublishSeqNo = channel.getNextPublishSeqNo();
            outstandmMap.put(nextPublishSeqNo,message);
        }
        long end = System.currentTimeMillis();
        // 异步确认发布 1000 条消息的耗时为 97 ms
        System.out.println("异步确认发布 " + MESSAGE_NUM + " 条消息的耗时为 "+ (end - start) +" ms");
    }
}
